package tpc3;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.net.SocketException;
import java.util.ArrayList;

import static commsCPN.EncodeDecode.*;
import commsCPN.JavaCPN;

public class Coordinator {
    private static Integer numOfWorkers = null;
    private static JavaCPN jCPN = new JavaCPN();
    private static STATES state = STATES.INIT;
    private static String[] received;

    public enum STATES {
        INIT, EVALUATE_VOTES, RECEIVE_ACKS;
        public STATES getNext() {
            return this.ordinal() < STATES.values().length - 1 ? STATES.values()[this.ordinal() + 1] : STATES.INIT;
        }
    };

    public static void main(String[] args) throws IOException, InterruptedException {
        System.out.println("Coordinator running.");
        CollectVotesLogic cVL = null;
        ArrayList<Integer> waitAcks = new ArrayList<Integer>();
        
        while (true) {
            switch (state) {
            case INIT:
                try {
                    jCPN.connect("localhost", 9000);
                    if (receive() == null)
                        break;
                    if (received[0].equals("numOfWorkers")) {
                        numOfWorkers = Integer.parseInt(received[received.length - 1]);
                        sendCanCommitMessages(numOfWorkers);
                        state = state.getNext();
                        System.out.println("-> state: EVALUATE_VOTES");
                    }
                } catch (java.net.ConnectException e) {
                    Thread.sleep(100);
                }
                break;
            case EVALUATE_VOTES:
                cVL = new CollectVotesLogic(jCPN, numOfWorkers);
                waitAcks.addAll(cVL.run());
                state = state.getNext();
                System.out.println("-> state: RECEIVE_ACKS");
                break;
            case RECEIVE_ACKS:
                //receive the expected acks, reset state.
                while (waitAcks.size()>0) {
                    if (receive() == null)
                        break;
                    if (received[0].equals("Acknowledge")){
                        Integer workerId = Integer.valueOf(Integer.parseInt(received[1].replaceAll("[^0-9]", "")));
                        if (waitAcks.contains(workerId)){
                            waitAcks.remove(workerId);
                        }
                    }
                }
                jCPN.disconnect();
                System.out.println("-> state: INIT (Cycle completed)");
                state = state.getNext();
                break;
            }
        }
    }

    private static void sendCanCommitMessages(int numOfWorkers) {
        StringBuilder sb = new StringBuilder();
        sb.append("CanCommit:");
        for (int i = 0; i < numOfWorkers; i++)
            sb.append(i).append(';');
        try {
            System.out.println("Sending: " + sb.toString());
            jCPN.send(encode(sb.toString()));
        } catch (SocketException e) {
            e.printStackTrace();
        }
    }

    private static String[] receive() {
        try {
            String rc = decodeString(jCPN.receive());
            received = rc.split(":");
            System.out.println("Received: " + rc);
            return received;
        } catch (java.net.SocketException e) {
            state = STATES.INIT;
            return null;
        }
    }
}
